package org.apache.zookeeper.server.paxos2;

import org.apache.jute.InputArchive;
import org.apache.jute.OutputArchive;
import org.apache.log4j.Logger;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.util.LinkedList;
import java.util.*;

public class Shard {

    private RandomAccessFile f;
    private static final Logger LOG = Logger.getLogger(Shard.class);
    private int stripeId;

    //private long[] versions;

    public Shard(int stripeId) {
        System.out.println("Initializing " + stripeId);
        this.stripeId = stripeId;
        try {
            f = new RandomAccessFile(Configuration.storagePath + File.separator + "storage", "rw");
            long length = Configuration.totalSize / Configuration.noOfStripes * Configuration.writeQuorumSize;
            if (f.length() < length) {
                LOG.info("file size does not match. create an empty file with size=" + length);
                f.setLength(length);
                byte[] tmp = new byte[Configuration.blockSize];
                for (int i = 0; i < (length / Configuration.blockSize); i++)
                    f.write(tmp);
            }
            /*versions = new long[(int) (length / Configuration.blockSize)];
            for (int i = 0; i < versions.length; i++)
                versions[i] = 0;*/
        }
        catch (Exception e) {
            LOG.error("Couldn't create storage file", e);
            System.exit(1);
        }
    }

    long totalTime = 0;
    int count = 0;

    /*private TreeMap<Long, byte[]> pendings = new TreeMap<Long, byte[]>();

    public void write(long blockId, byte[] data, long version){

	pendings.put(blockId, data);
	if(pendings.size()<100)
	    return;
	for(Long tmp:pendings.keySet())
	    write2(tmp, pendings.get(tmp), -1);
	pendings.clear();
    }*/

    public void write(long blockId, byte[] data) {
        //if(true) return;
        long offset = Mapping.keyToOffsetInShard(blockId);
        long blockIdInShard = blockId / Configuration.noOfStripes;
	//System.out.println("write "+blockId+" "+data[0]+""+data[1]+""+data[2]+""+data[3]+" length="+data.length);
	if(blockIdInShard>Configuration.totalSize / Configuration.noOfStripes/Configuration.blockSize)
	    throw new RuntimeException("Cannot write to "+blockIdInShard+" max is "+Configuration.totalSize / Configuration.noOfStripes/Configuration.blockSize);
        //System.out.println("write blockId "+blockId+" to "+stripeId+" "+offset);
        try {
            long tmp = System.currentTimeMillis();
            f.seek(offset);
            f.write(data);
            long time = System.currentTimeMillis() - tmp;
            if (time > 1000)
                System.out.println("long write " + time);
	    //if(count%1000==999)
	//	f.getFD().sync();
            /*totalTime += time;
            count++;
            if (count % 100 == 99) {
                System.out.println(stripeId + " average = " + ((double) totalTime / (double) count));
                //	count=0;
                //	totalTime=0;
            }*/
	    /*if(Runtime.getRuntime().freeMemory()<Runtime.getRuntime().maxMemory()/2){
		System.out.println("gc");
		System.gc();
	    }*/
        }
        catch (Exception e) {
            LOG.error("Couldn't write blockId=" + blockId, e);
            System.exit(1);
        }
    }

    public void read(long blockId, byte[] data) {
        /*if(blockId == 301){
          System.out.println("Deliberate killing");
          System.exit(0);
      }*/
        long offset = Mapping.keyToOffsetInShard(blockId);
	//System.out.println(stripeId+" read="+blockId+" offset="+offset);
        try {
            f.seek(offset);
            f.readFully(data);
	    //System.out.println("read "+blockId+" "+data[0]+""+data[1]+""+data[2]+""+data[3]);
        }
        catch (Exception e) {
            LOG.error("Couldn't read blockId=" + blockId, e);
            System.exit(1);
        }
    }

    public void flush() {
        try {
            f.getFD().sync();
        }
        catch (Exception e) {
            LOG.error("Couldn't flush", e);
            System.exit(1);
        }
    }

    /*public void getNewBlocks(long zxid, List<Long> blocks, List<Long> blockVersions) {
        for (int i = 0; i < versions.length; i++) {
            if (versions[i] > zxid) {
                blocks.add(Mapping.getBlock(this.stripeId, i));
                blockVersions.add(versions[i]);
            } else if (versions[i] != 0 && Configuration.debug)
                System.out.println("versions" + i + "=" + versions[i]);
        }
        System.out.println(this + " Shard" + this.stripeId + " has " + blocks.size() + " new blocks zxid=" + zxid);
    } */

    public void serialize(OutputArchive oa, String tag) throws IOException {
        /*System.out.println("serialize shard " + stripeId);
        for (int i = 0; i < versions.length; i++)
            oa.writeLong(versions[i], "version");*/
        flush();
    }

    public void deserialize(InputArchive ia, String tag) throws IOException {
        /*for (int i = 0; i < versions.length; i++)
            versions[i] = ia.readLong("version");
        System.out.println("deserialize versions " + versions.length)*/;
    }

    public void close() {
        try {
            f.close();
        }
        catch (Exception e) {
            LOG.error("Couldn't close", e);
        }
    }
}
